package studio.raptor.databus.client;


import studio.raptor.databus.ActiveMQConnectionFactory;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;

/**
 *
 * 消费者配置，包含消费者所属业务中心标志，消费的主题及消费者个性化配置。<br><br>
 *
 * <P>
 *     消费者默认配置如下：<br>
 *     会话不开启事务；<br>
 *     消费者自动ACK消息模式；<br>
 *     开启批量ACK功能；<br>
 *     批量ACK超时时间30s；<br>
 *     消息重发次数不受限制；<br>
 *     每个消费者预取消息数量200；<br>
 *     两次重连之间最长的时间间隔为10s(从10毫秒开始指数增加到10s)<br>
 *     failover后的尝试重连次数<br>
 * </P>
 */
public final class DatabusConsumerBuilder {
    /**
     * 业务中心标志
     */
    private String bizCenterKey;
    /**
     * 主题
     */
    private String topic;

    /**
     * 会话是否为事务性，消费者默认为否
     */
    private boolean transactedSession = false;

    /**
     * 消息ACK模式，默认值(1)为自动确认
     */
    private int acknowledgeMode = Session.AUTO_ACKNOWLEDGE ;
    /**
     * 开启批量ACK优化
     */
    private boolean optimizeAcknowledge = true;
    /**
     * ACK消息超时时间，默认30s
     */
    private int optimizeAcknowledgeTimeOut = 30000;
    /**
     * 消息重发次数，默认5，表示直到消息被ACK，否则一直重发。0表示不重传,-1表示一直重发
     */
    private int maximumRedeliveries = 5;
    /**
     * 默认预取值设定为200
     */
    private int prefetchSize = 200 ;
    /**
     * 两次重连之间最长时间间隔，10s.
     */
    private int maxReconnectDelay = 10000;
    /**
     * failover机制中的最大重连次数。-1表示一直重连。
     */
    private int maxReconnectAttempts = -1;

    /**
     * 客户端启动时最大重连次数，默认为5
     */
    private int startupMaxReconnectAttempts = 5;
    /**
     * 通用其他配置项
     */
    private String otherURLConfig ="";

    /**
     * 死信队列名称前缀,默认为<code>DLQ.</code>
     */
    private String DLQPre = "DLQ.";

    private DatabusConfig config;


    public DatabusConsumerBuilder(String bizCenterKey, String topic,DatabusConfig config) {
        this.bizCenterKey = bizCenterKey;
        this.topic = topic;
        this.config = config;
    }

    /**
     * 创建死信队列消费者
     * @return
     * @throws JMSException
     */
    public DatabusConsumer createDLQ() throws JMSException{
        // 获取消费者客户端对象
        DatabusConsumer databusConsumer = new DatabusConsumer();

        // 配置连接工厂类
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setUserName(config.getUserName());
        connectionFactory.setPassword(config.getPassword());
        String customizedBrokerURL = this.transform2URLConfig();
        connectionFactory.setBrokerURL(config.getBrokerURL()+"?"+customizedBrokerURL);

        // 配置消费者客户端中的连接、会话
        databusConsumer.setConnection(connectionFactory.createConnection());
        databusConsumer.getConnection().start();
        databusConsumer.setSession(databusConsumer.getConnection().createSession(this.isTransactedSession(),this.getAcknowledgeMode()));

        // 创建消费者客户端中的原始消费者
        String destinationURLConfig = this.transform2DestinationConfig();
        String groupQueueName =
                this.getDLQPre()+config.GROUP_QUEUE_NAME_PRE+
                        this.getBizCenterKey()+"."+
                        config.VIRTUAL_TOPIC_PRE+this.getTopic();
        Destination destination =
                databusConsumer.getSession().createQueue(groupQueueName+"?"+destinationURLConfig);
        databusConsumer.setMessageConsumer(databusConsumer.getSession().createConsumer(destination));

        return databusConsumer ;
    }

    public DatabusConsumer create() throws JMSException {
        // 获取消费者客户端对象
        DatabusConsumer databusConsumer = new DatabusConsumer();

        // 配置连接工厂类
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setUserName(config.getUserName());
        connectionFactory.setPassword(config.getPassword());
        String customizedBrokerURL = this.transform2URLConfig();
        connectionFactory.setBrokerURL(config.getBrokerURL()+"?"+customizedBrokerURL);

        // 配置消费者客户端中的连接、会话
        databusConsumer.setConnection(connectionFactory.createConnection());
        databusConsumer.getConnection().start();
        databusConsumer.setSession(databusConsumer.getConnection().createSession(this.isTransactedSession(),this.getAcknowledgeMode()));

        // 创建消费者客户端中的原始消费者
        String destinationURLConfig = this.transform2DestinationConfig();
        String groupQueueName =
                config.GROUP_QUEUE_NAME_PRE+
                        this.getBizCenterKey()+"."+
                        config.VIRTUAL_TOPIC_PRE+this.getTopic();
        Destination destination =
                databusConsumer.getSession().createQueue(groupQueueName+"?"+destinationURLConfig);
        databusConsumer.setMessageConsumer(databusConsumer.getSession().createConsumer(destination));

        return databusConsumer ;

    }


    public String getBizCenterKey() {
        return bizCenterKey;
    }

    public void setBizCenterKey(String bizCenterKey) {
        this.bizCenterKey = bizCenterKey;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }


    public boolean isTransactedSession() {
        return transactedSession;
    }

    /**
     *  设定会话是否为事务性，消费者默认为否
     * @param transactedSession
     */
    public void setTransactedSession(boolean transactedSession) {
        this.transactedSession = transactedSession;
    }

    public int getAcknowledgeMode() {
        return acknowledgeMode;
    }

    /**
     * 消息ACK模式
     * @param acknowledgeMode 默认(1)为自动确认模式。2为手工批量ACK，4为手工单条消息确认。
     */
    public void setAcknowledgeMode(int acknowledgeMode) {
        this.acknowledgeMode = acknowledgeMode;
    }

    public boolean isOptimizeAcknowledge() {
        return optimizeAcknowledge;
    }

    /**
     * 设置是否开启批量ACK操作(当且仅当session不是事务性且ACK模式为自动确认时，开启此功能才有意义)
     * @param optimizeAcknowledge 默认开启
     */
    public void setOptimizeAcknowledge(boolean optimizeAcknowledge) {
        this.optimizeAcknowledge = optimizeAcknowledge;
    }

    public int getOptimizeAcknowledgeTimeOut() {
        return optimizeAcknowledgeTimeOut;
    }

    /**
     * 设置批量ACK超时时间(当且仅当session不是事务性且ACK模式为自动确认时，此值才有意义)
     * @param optimizeAcknowledgeTimeOut ACK超时时间，单位毫秒
     */
    public void setOptimizeAcknowledgeTimeOut(int optimizeAcknowledgeTimeOut) {
        this.optimizeAcknowledgeTimeOut = optimizeAcknowledgeTimeOut;
    }

    public int getMaximumRedeliveries() {
        return maximumRedeliveries;
    }

    /**
     * 设置消息最大重发次数(建议使用默认值)
     * @param maximumRedeliveries 默认为-1，表示一直会重发
     */
    public void setMaximumRedeliveries(int maximumRedeliveries) {
        this.maximumRedeliveries = maximumRedeliveries;
    }

    public int getPrefetchSize() {
        return prefetchSize;
    }

    /**
     * 设置消费者的预取消息数量
     * @param prefetchSize 默认200
     */
    public void setPrefetchSize(int prefetchSize) {
        this.prefetchSize = prefetchSize;
    }

    public int getMaxReconnectDelay() {
        return maxReconnectDelay;
    }

    /**
     * 设置最大重来延迟毫秒数
     * @param maxReconnectDelay 默认10s
     */
    public void setMaxReconnectDelay(int maxReconnectDelay) {
        this.maxReconnectDelay = maxReconnectDelay;
    }

    public int getMaxReconnectAttempts() {
        return maxReconnectAttempts;
    }

    /**
     * 设置failover机制最大重连次数
     * @param maxReconnectAttempts 默认值为-1，表示一直尝试重连。0表示不重连
     */
    public void setMaxReconnectAttempts(int maxReconnectAttempts) {
        this.maxReconnectAttempts = maxReconnectAttempts;
    }

    public String getOtherURLConfig() {
        return otherURLConfig;
    }

    /**
     * 设置其他参数，默认情况下不使用此项
     * @param otherURLConfig
     */
    public void setOtherURLConfig(String otherURLConfig) {
        this.otherURLConfig = otherURLConfig;
    }


    public String getDLQPre() {
        return DLQPre;
    }

    /**
     * 设置死信队列名称的前缀，默认为 "DLQ."，不建议修改
     * @param DLQPre 死信队列默认前缀
     */
    public void setDLQPre(String DLQPre) {
        this.DLQPre = DLQPre;
    }

    public int getStartupMaxReconnectAttempts() {
        return startupMaxReconnectAttempts;
    }

    /**
     * 设置客户端启动时尝试连接服务器的最大次数，默认为5
     * @param startupMaxReconnectAttempts  客户端启动时尝试连接服务器的最大次数
     */
    public void setStartupMaxReconnectAttempts(int startupMaxReconnectAttempts) {
        this.startupMaxReconnectAttempts = startupMaxReconnectAttempts;
    }

    /**
     * 将消费者客户端配置转换为URL参数
     * @return
     */
    public String transform2URLConfig(){
        StringBuilder urlConfig = new StringBuilder();
        urlConfig.append("jms.optimizeAcknowledge="+this.isOptimizeAcknowledge());
        urlConfig.append("&jms.optimizeAcknowledgeTimeOut="+this.getOptimizeAcknowledgeTimeOut());
        urlConfig.append("&jms.redeliveryPolicy.maximumRedeliveries="+this.getMaximumRedeliveries());
        urlConfig.append("&maxReconnectDelay="+this.getMaxReconnectDelay());
        urlConfig.append("&maxReconnectAttempts="+this.getMaxReconnectAttempts());
        urlConfig.append("&startupMaxReconnectAttempts="+this.getStartupMaxReconnectAttempts());
        urlConfig.append(this.getOtherURLConfig());
        return urlConfig.toString();
    }

    /**
     * 将消费者的消息目的地配置转换
     * @return
     */
    public String transform2DestinationConfig(){
        StringBuilder destinationURLConfig = new StringBuilder();
        destinationURLConfig.append("consumer.prefetchSize="+this.getPrefetchSize());
        return  destinationURLConfig.toString();
    }
}
